package com.tyc.threalpool;

import java.util.ArrayList;
import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;

/**
 * kafka消费者拉取到一批数据，解析，批量保存到mysql
 * 若其中一条sql异常导致这批数据丢失，解决方案如下
 * 若想在异常后重新提交任务：继承ThreadPoolExecutor重写afterExecute
 */
public class Demo02 {
    public static void main(String[] args) {
        ExecutorService executor = Executors.newFixedThreadPool(5,new MyThreadFactory());
        // list模拟这一批数据
        List<String> list = new ArrayList<>();
        list.add("123");
        list.add("122");
        list.add("133");
        list.add("aaa");
        list.add("111");
        executor.execute(()->{
            boolean error = true;
            String errorItem = null;
            try {
                for (String s : list) {
                    errorItem = s;
                    Integer.parseInt(s); // 模拟一条sql执行数据
                }
                System.out.println("刷到数据库");
                error = false;
            } catch (NumberFormatException e) {
                throw e;
            } finally {
                if(error){
                    System.out.println("报警：数据{"+errorItem+"}处理异常");
                    for (String s : list) {
                        // 将日志输出到data里，后续进行补数据
                        System.out.println(s);
                    }
                }
            }
        });
    }

    static class MyThreadFactory implements ThreadFactory {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("my-");
            thread.setUncaughtExceptionHandler((t,e)->{
                System.out.println("捕获到异常");
            });
            return thread;
        }
    }
}
